"""
Case Type   : 资源池化支持xlog归档
Case Name   : 设置资源池化主机xlog归档GUC参数，测试PITR功能
Create At   : 2025/06/07
Owner       : @hwworkholic
Description :
    1.配置归档参数
    2.预置数据
    3.执行备份
    4.预置增量恢复数据，创建restart point
    5.测试PITR
Expect      :
    1.成功
    2.成功
    3.成功
    4.成功
    5.成功
History     :
"""

import os
import subprocess
import unittest

from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger
from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH

from yat.test import Node
from yat.test import macro


@unittest.skipIf(
    "dms_res" not in Common().get_sh_result(
        Node("PrimaryDbUser"),
        f"source {macro.DB_ENV_PATH};cm_ctl query -Cvidp"), "非资源池化不执行")
class SharedStorageAchiveXlog(unittest.TestCase):
    def setUp(self):
        self.logger = Logger()
        self.logger.info(f"----{os.path.basename(__file__)} start----")
        self.constant = Constant()
        self.com = Common()
        self.primary_sh = CommonSH("PrimaryDbUser")
        self.primary_node = Node("PrimaryDbUser")
        self.table_name = "t_archive_xlog_test018"
        self.table_name_1 = "t_archive_xlog_test018_01"
        self.table_name_2 = "t_archive_xlog_test018_02"
        self.table_name_3 = "t_archive_xlog_test018_03"
        self.archive_dest = os.path.join(macro.DB_INSTANCE_PATH,
                                         "archive_dir/")
        self.default_archive_mode = \
            self.com.show_param("archive_mode")
        self.default_archive_dest = \
            self.com.show_param("archive_dest")
        self.archive_command = 'dsscmd cp -s %p -d %f'
        self.parent_path = os.path.dirname(macro.DB_INSTANCE_PATH)
        self.backup_path = os.path.join(self.parent_path,
                                        "backupset/backup_dir_018")
        self.instance_name = 'probackup_018'
        self.recover_conf = \
            os.path.join(macro.DB_INSTANCE_PATH, 'recovery.conf')
        cmd = f'source {macro.DB_ENV_PATH};' \
              f'cat $DSS_HOME/cfg/dss_vg_conf.ini'
        self.vgname_infor = self.primary_node.sh(cmd).result().strip().split('\n')
        self.vglist = [line.split(':')[0] for line in self.vgname_infor]
        self.diskname = [line.split(':')[1] for line in self.vgname_infor]
        col1_with_plus = ['+' + item for item in self.vglist]
        self.vgname = ",".join(col1_with_plus)

    def test_archive_xlog(self):
        step = '-----step 1:检查集群初始状态; expect:集群状态正常-----'
        self.logger.info(step)
        status = self.primary_sh.exec_cm_ctl(mode='query', param='-Cv')
        self.logger.info(status)
        self.assertIn('cluster_state   : Normal', status, '执行失败' + step)

        step = "-----step 2:设置archive_mode=on expect:成功-----"
        self.logger.info(step)
        guc_res = self.primary_sh.execute_gsguc(
            "reload", self.constant.GSGUC_SUCCESS_MSG,
            f"archive_mode=on")
        self.logger.info(guc_res)
        self.assertTrue(guc_res, "执行失败:" + step)

        step = "-----step3:设置archive_command expect:成功-----"
        self.logger.info(step)
        guc_res = self.primary_sh.execute_gsguc(
            "reload", self.constant.GSGUC_SUCCESS_MSG,
            f"archive_command='{self.archive_command}'")
        self.logger.info(guc_res)
        self.assertTrue(guc_res, "执行失败:" + step)

        step = "-----step4:设置archive_dest expect:成功-----"
        self.logger.info(step)
        guc_res = self.primary_sh.execute_gsguc(
            "reload", self.constant.GSGUC_SUCCESS_MSG,
            f"archive_dest='{self.archive_dest}'")
        self.logger.info(guc_res)
        self.assertTrue(guc_res, "执行失败:" + step)

        step = "--step 5:创建基础表，插入数据 expect:成功--"
        self.logger.info(step)
        sql_cmd = f"drop table if exists {self.table_name};" \
                  f"create table {self.table_name}(id int,name char(20));" \
                  f"insert into {self.table_name} values " \
                  f"(generate_series(1,1000), 'test');"
        self.logger.info(sql_cmd)
        sql_res = self.primary_sh.execut_db_sql(sql_cmd)
        self.logger.info(sql_res)
        self.assertIn(self.constant.INSERT_SUCCESS_MSG, sql_res,
                      "执行失败:" + step)

        step = "--step 6:执行全量备份 expect:成功--"
        self.logger.info(step)
        self.logger.info("-----初始化目录-----")
        result = self.primary_sh.exec_pro_backup_init(self.backup_path)
        self.logger.info(result)
        self.assertTrue(result, "执行失败:" + step)
        self.logger.info("-----添加备份实例-----")
        result = self.primary_sh.exec_pro_back_add(
            self.backup_path, self.instance_name,
            f"--enable-dss --vgname='{self.vgname}'")
        self.logger.info(result)
        self.assertTrue(result, "执行失败:" + step)
        self.logger.info("-----执行全量备份-----")
        result = self.primary_sh.exec_pro_backup_backup(
            self.backup_path, self.instance_name, 'full',
            f"{self.primary_node.db_name}",
            f"--vgname='{self.vgname}' -j 128", True)
        self.assertIn('completed', result, "执行失败:" + step)
        self.backup_id = ''
        for backupid_msg in result.splitlines():
            if 'completed' in backupid_msg:
                self.backup_id = backupid_msg.split()[2]
        self.logger.info('备份ID为：' + self.backup_id)

        step = "--step 7:多次创建表，插入数据，并记录point expect:成功--"
        self.logger.info(step)
        self.logger.info("--第一次创建表1，记录point1--")
        sql_cmd = f"drop table if exists {self.table_name_1};" \
                  f"create table {self.table_name_1}(id int,name char(20));" \
                  f"insert into {self.table_name_1} values " \
                  f"(generate_series(1,1000), 'test');"
        self.logger.info(sql_cmd)
        sql_res = self.primary_sh.execut_db_sql(sql_cmd)
        self.logger.info(sql_res)
        self.assertIn(self.constant.INSERT_SUCCESS_MSG, sql_res,
                      "执行失败:" + step)
        point1_res = self.primary_sh.execut_db_sql(
            "select pg_create_restore_point('point1');")
        self.logger.info(point1_res)
        self.logger.info("restore point 是：point1")

        self.logger.info("--第二次创建表2，记录point2--")
        sql_cmd = f"drop table if exists {self.table_name_2};" \
                  f"create table {self.table_name_2}(id int,name char(20));" \
                  f"insert into {self.table_name_2} values " \
                  f"(generate_series(1,1000), 'test');"
        self.logger.info(sql_cmd)
        sql_res = self.primary_sh.execut_db_sql(sql_cmd)
        self.logger.info(sql_res)
        self.assertIn(self.constant.INSERT_SUCCESS_MSG, sql_res,
                      "执行失败:" + step)
        point2_res = self.primary_sh.execut_db_sql(
            "select pg_create_restore_point('point2');")
        self.logger.info(point2_res)
        self.logger.info("restore point 是：point2")

        self.logger.info("--第三次创建表3，记录point3--")
        sql_cmd = f"drop table if exists {self.table_name_3};" \
                  f"create table {self.table_name_3}(id int,name char(20));" \
                  f"insert into {self.table_name_3} values " \
                  f"(generate_series(1,1000), 'test');"
        self.logger.info(sql_cmd)
        sql_res = self.primary_sh.execut_db_sql(sql_cmd)
        self.logger.info(sql_res)
        self.assertIn(self.constant.INSERT_SUCCESS_MSG, sql_res,
                      "执行失败:" + step)
        point3_res = self.primary_sh.execut_db_sql(
            "select pg_create_restore_point('point3');")
        self.logger.info(point3_res)
        self.logger.info("restore point 是：point3")

        step = "--step 8:执行pg_switch_xlog expect:成功--"
        self.logger.info(step)
        archive_res = self.primary_sh.execut_db_sql("select pg_switch_xlog();")
        self.logger.info(archive_res)
        self.assertIn("1 row", archive_res, "执行失败:" + step)

        step = "--step 9:使用归档日志恢复到recovery_target_name expect:成功--"
        self.logger.info(step)
        result = self.primary_sh.exec_cm_ctl('stop')
        self.logger.info(result)
        self.assertIn(self.constant.cm_stop_success_msg, result,
                      "停止集群失败")
        self.logger.info("-----执行全量恢复: dd清理磁盘-----")
        for item in self.diskname:
            cmd = f'dd if=/dev/zero of={item} bs=2048 count=1000 > /dev/null 2>&1'
            result = self.primary_node.sh(cmd).result()
            self.logger.info(result)
        self.logger.info("-----执行全量恢复: 重建卷-----")
        for item1, item2 in zip(self.vglist, self.diskname):
            cmd = f'source {macro.DB_ENV_PATH};dsscmd cv -g {item1} -v {item2} -D $DSS_HOME;'
            result = self.primary_node.sh(cmd).result()
            self.logger.info(result)
        self.logger.info("-----执行全量恢复: 删除pgdata-----")
        cmd = f'source {macro.DB_ENV_PATH};rm -rf $PGDATA;'
        result = self.primary_node.sh(cmd).result()
        self.logger.info(result)
        self.logger.info("-----执行全量恢复: 启动dss-----")
        cmd = f'source {macro.DB_ENV_PATH};dssserver -M -D $DSS_HOME > $DSS_HOME/startdss.log 2>&1 &;'
        result = self.primary_node.sh(cmd).result()
        self.logger.info(result)
        result = self.primary_sh.exec_pro_backup_restore(
            self.backup_path, self.instance_name, self.backup_id,
            restore_cmd=f'-D {macro.DB_INSTANCE_PATH} -j 128')
        self.logger.info(result)
        self.assertTrue(result, "主机执行恢复失败")
        self.logger.info('----停止dssserver----')
        stop_dss_cmd = f"source {macro.DB_ENV_PATH};dsscmd stopdss;"
        res = self.com.get_sh_result(self.primary_node, stop_dss_cmd)
        self.logger.info(res)
        self.assertIn('Succeed to stop server.', res, "dss停止失败")
        self.logger.info("----配置recovery.conf文件----")
        cmd = f"touch " \
              f"{os.path.join(macro.DB_INSTANCE_PATH, 'recovery.conf')};" \
              f"echo \"restore_command = 'dsscmd cp -s " \
              f"{self.archive_dest}%f -d %p'\" > " \
              f"{self.recover_conf};" \
              f"echo \"recovery_target_name   = 'point2'\" >> " \
              f"{self.recover_conf};" \
              f"echo \"recovery_target_inclusive = false\" >> " \
              f"{self.recover_conf};" \
              f"echo \"pause_at_recovery_target = false\" >> " \
              f"{self.recover_conf};" \
              f"cat {self.recover_conf}"
        self.logger.info(cmd)
        result = self.primary_node.sh(cmd).result()
        self.logger.info(result)
        self.logger.info("----启动集群----")
        start_res = self.primary_sh.exec_cm_ctl('start')
        self.logger.info(start_res)
        self.assertIn(self.constant.cm_start_success_msg, start_res,
                      "启动集群失败")

        step = "--step 10:查询恢复后的数据 expect:恢复后的数据与point节点前的数据一致--"
        self.logger.info(step)
        sql_cmd = f"select count(*) from {self.table_name};" \
                  f"select count(*) from {self.table_name_1};" \
                  f"select count(*) from {self.table_name_2};" \
                  f"select count(*) from {self.table_name_3};"
        self.logger.info(sql_cmd)
        sql_res = self.primary_sh.execut_db_sql(sql_cmd)
        self.logger.info(sql_res)
        self.assertEqual(4, sql_res.count("1000"), "执行失败:" + step)

    def tearDown(self):
        self.logger.info("----清理环境----")
        rm_cmd = f"rm -rf {self.archive_dest};" \
                 f"rm -rf {self.backup_path};" \
                 f"rm -rf {self.recover_conf}"
        self.logger.info(rm_cmd)
        rm_msg = self.primary_node.sh(rm_cmd).result()
        drop_res = self.primary_sh.execut_db_sql(
            f"drop table if exists {self.table_name},{self.table_name_1},"
            f"{self.table_name_2},{self.table_name_3};")
        self.logger.info(drop_res)
        guc_res1 = self.primary_sh.execute_gsguc(
            "reload", self.constant.GSGUC_SUCCESS_MSG,
            f"archive_mode={self.default_archive_mode}")
        self.logger.info(guc_res1)
        guc_res2 = self.primary_sh.execute_gsguc(
            "reload", self.constant.GSGUC_SUCCESS_MSG,
            f"archive_dest='{self.default_archive_dest}'")
        self.logger.info(guc_res2)
        self.assertTrue(drop_res, "删除表失败")
        self.assertEqual("", rm_msg, "删除文件失败")
        self.logger.info("----恢复集群----")
        self.logger.info(f'----{os.path.basename(__file__)} end----')
